package com.flinkjar.flinkjardemo.common;


import com.alibaba.fastjson.JSONObject;
import com.flinkjar.flinkjardemo.bean.UserProduct;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Struct;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.source.SourceRecord;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;


/**
 * 自定义的 DebeziumDeserializationSchema 实现
 */
public class CustomDeserializationSchema implements DebeziumDeserializationSchema<UserProduct> {

    /**
     * 日志
     */
    private static final Logger logger = LoggerFactory.getLogger(CustomDeserializationSchema.class);


    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<UserProduct> collector) throws Exception {
        // 获取 database和 table
        String topic = sourceRecord.topic();
        // 分隔符得写 \\. 不然就报错
        String[] strings = topic.split("\\.");
        String database = strings[1];
        String table = strings[2];
        //logger.info("转换数据-数据库={}, 数据表={}", database, table);

        // 获取操作类型
        Envelope.Operation op = Envelope.operationFor(sourceRecord);
        //logger.info("转换数据-操作类型={}", op);

        // 获取 data
        // eg: Struct{after=Struct{id=146,user_id=6183cc27439df885e0c015da,product_id=5f6463674407101fafa9f708,product_spec_id=6322c1907385a03c11ab3fcc},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1712126137816,snapshot=true,db=flink_test_db,table=user_product,server_id=0,file=binlog.000071,pos=2417373,row=0},op=r,ts_ms=1712126137819}
        Struct value = (Struct) sourceRecord.value();

        // 变更之前值
        Struct before = value.getStruct(Envelope.FieldName.BEFORE);
        //logger.info("转换数据-before={}", before);

        // 变更之后值
        Struct after = value.getStruct(Envelope.FieldName.AFTER);
        //logger.info("转换数据-after={}", after);


        Struct finalStruct = after;
        if (Envelope.Operation.DELETE.equals(op)) {
            // 删除数据, after=null, 取before
            finalStruct = before;
        }

        // 除删除数据以外, 只要after的数据(before是变更前的数据不需要)
        UserProduct userProduct = new UserProduct();
        userProduct.setOperation(op);
        // 针对无符号的 bigint 需要使用BigDecimal接收.
        BigDecimal decimal = (BigDecimal) finalStruct.get("id");
        userProduct.setId(Long.parseLong(decimal.toString()));

        userProduct.setUserId(finalStruct.getString("user_id"));
        userProduct.setProductId(finalStruct.getString("product_id"));

        //logger.info("转换数据-成功, data={}", JSONObject.toJSONString(userProduct));
        // 采集数据
        collector.collect(userProduct);
    }


    /**
     * 定义返回的类型
     *
     * @return
     */
    @Override
    public TypeInformation<UserProduct> getProducedType() {
        return TypeInformation.of(UserProduct.class);
    }

}
